
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：ssdb封装 跟具体业务无关
--- 各业务服务器直连ssdb数据库的时候使用，
--- 使用ngx自带的连接池，长连接到ssdb数据库，
--- 业务服务器 --(ssdbMgr.lua)--> ssdb数据库
----------------------------------------------------------------
--- 本文件实现 业务服直连ssdb数据库，是比较原始的做法，
--- 新做法是使用data_cache作为共享内存缓存中间层，性能更好，
--- 请参考 commhm/dc/ssdbMgrDc.lua
----------------------------------------------------------------

local assert     = assert
local require    = require
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

require "resty.core.shdict"

local ngx_shared  = ngx.shared;
local string_sub  = string.sub
local string_gsub = string.gsub
local string_format = string.format

local log       = require("modules.commhm.log");
local ns_time   = require("modules.commhm.time");
local uu        = require("modules.commhm.utils");

local ns_ssdb   = require("modules.commhm.resty.ssdb");

local ns_semaphore  = require("ngx.semaphore")
--local ns_lock     = require("modules.commhm.resty.lock")


local table_insert = table.insert
local ngx_sleep = ngx.sleep
local ngx_now   = ngx.now


local _SSDB_CALL_LOG_  = false    --日志

local _SSDB_RESTY_LOCK      = true     --shared lock

local __TJ_INTERVAL__      = 180
local __SSDB_DELAY__       = 60    --每60秒写入一次


-- SSDB MGR类
local _M = { _VERSION = '0.12.1' }
local mt = { __index = _M }


function _M.new(self, pre_, config_server_, config_select_ )
	log.debug( "call ssdbMgr:new" );
	local ins_ = {
		config_cache  = {};
		pre           = pre_;

		config_server = config_server_;
		config_select = config_select_;

		read_semas_tj = {};

		last_time_report_tj        = uu.now() + __TJ_INTERVAL__;

		get_set_tj   = {};
		-- get_call
		-- get_cache
		-- get_cache_stale
		-- get_ssdb
		-- set_ssdb
		-- delay_ssdb

		mread_tj = {
			call  = 0,
			get   = 0,
			set   = 0,
		};

	};
    return setmetatable(ins_, mt);
end


_M.init_pool = function(self)
end


---处理统计，关闭resty_locks
_M.final = function(self)
	--释放locks
	if  ngx.ctx.resty_locks then
		for k, v in pairs( ngx.ctx.resty_locks ) do
			if  v and v.dict then
				if  v.cc > 0 then
					local ok, err_ = v.dict:delete( k .. "_lk" )
					if  ok then
						if  v.cc > 1 then
							log.day_list_simple( "lock_rw", "unlock_ok2", ngx.worker.id(), uu.get_act_cmd(), k, v.cc, debug.traceback() );
						end
					else
						log.day_list_simple( "lock_rw", "error_unlock2", ngx.worker.id(), uu.get_act_cmd(), k, v.cc, err_, debug.traceback() );
					end
				else
					--log.day_list_simple( "lock_rw", "unlocked2", ngx.worker.id(), uu.get_act_cmd(), k, v.cc, debug.traceback() );
				end
			else
				log.day_list_simple( "lock_rw", "error_locks2", ngx.worker.id(), uu.get_act_cmd(), k, v.cc, debug.traceback() );
			end
		end
		ngx.ctx.resty_locks = nil
	end
	
	
end



_M.set_get_set_tj = function(self, hash_, name_)
	if     ngx.ctx.m_params.act then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.act
	elseif ngx.ctx.m_params.cmd then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.cmd
	end

	if  not self.get_set_tj[hash_] then
		self.get_set_tj[hash_] = {
			get_call  = 0,
			get_cache = 0,
			get_cache_stale  = 0,
			get_ssdb = 0,
			set_ssdb = 0,
			delay_ssdb = 0,
		};
	end

	if  self.get_set_tj[hash_][ name_ ] then
		self.get_set_tj[hash_][ name_ ] = self.get_set_tj[hash_][ name_ ] + 1
	else
		log.error( "call set_get_set_tj, unknown name_=" .. name_ )
	end

end


_M.do_report_tj = function(self)
	local now_ = uu.now()
	if  now_ > self.last_time_report_tj then
		self.last_time_report_tj = now_ + __TJ_INTERVAL__;

		--开始打印统计
		for hash_, v1 in pairs(self.get_set_tj) do
			if  v1.get_call + v1.set_ssdb > 0 then
				log.day_list_short( "cache_hit_ssdb", "get_set", self.pre,
					                "get", v1.get_call, v1.get_cache,  v1.get_cache_stale, v1.get_ssdb,
									"set", v1.set_ssdb, v1.delay_ssdb, hash_ )
			end
		end
		self.get_set_tj = {}

		if  self.mread_tj.call + self.mread_tj.get + self.mread_tj.set > 0 then
			log.day_list_short( "cache_hit_ssdb", "mread", self.pre, self.mread_tj.call, self.mread_tj.get, self.mread_tj.set )
			self.mread_tj = { call=0, get=0, set=0 }
		end
	end
end


-------------------------------------------
--kvx 独立库
-------------------------------------------
_M.set_kv = function( self, ssdb_name, key, value, et, err_call_back )
	--if  _SSDB_CALL_LOG_ then
		--log.day_list_short( "ssdb_call",  ngx.ctx.m_params.act, ngx.ctx.m_params.cmd, "setx", ssdb_name, key);
	--end

	log.debug( "call ssdbMgr.set_kv, ssdb_name=" .. ssdb_name .. ", key=" .. key );
	ngx.ctx.err_call_back = err_call_back;

	--优先写入share cache
	--if  false then
		--self:set_shared_cache_ssdb( ssdb_name, key .. "_kv", value, et )
	--end

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:read_unlock2(ssdb_name)
		self:error_exit( "write_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ok, err  = instance:setx(key, "" .. value, et);
	self:read_unlock2(ssdb_name)

	if  not ok then
		log.day( "ssdb",  "setx_fail|" .. ssdb_name .. "|" .. key .. "|" .. (err or "nil") );
		self:error_exit( "setx err:" .. (err or "nil") );
	else
		self:close()
	end

	self:set_get_set_tj( "kv", "set_ssdb" )
end


_M.get_kv = function( self, ssdb_name, key, err_call_back )
	--if  _SSDB_CALL_LOG_ then
		--log.day_list_short( "ssdb_call",  ngx.ctx.m_params.act, ngx.ctx.m_params.cmd, "get", ssdb_name, key );
	--end

	log.debug( "call ssdbMgr.get_kv, ssdb_name=" .. ssdb_name .. ", key=" .. key );
	ngx.ctx.err_call_back = err_call_back;

	--if  false then
		--local ret2, stale_ = self:get_shared_cache_ssdb( ssdb_name, key .. "_kv" )
		--if  ret2 then
			--if  stale_ then
				--log.debug( "ignore stale_ ret2kv cache hit." )
			--else
				--log.debug( "ret2kv cache hit." )
				--return ret2   --cache命中
			--end
		--end
	--end

	self:set_get_set_tj( "kv", "get_call" )

	local ret = ngx.null;

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:read_unlock2(ssdb_name)
		self:error_exit( "write_lock_err" );
		return
	end


	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:get(key);
	self:read_unlock2(ssdb_name)

	--local ttl_ = instance:ttl( key );
	--uu.var_dump( ttl_ )

	if  err then
		log.day( "ssdb", "get_fail|" .. key .. "|" .. (err or "nil") );
		self:error_exit( "get err:" .. (err or "nil") );
		return;
	else
		self:close()
	end

	uu.var_dump( ret_table_ );

	-- 处理返回值 返回是字符串或者ngx.null
	if  type(ret_table_) == 'table' then
		ret = ret_table_[1];
		if  ret and type(ret) == 'string' then
			if  'not_found' == ret then   --ssdb not_found
				--结果为空
				log.debug( "ssdb ret not_found" );
				ret = ngx.null;
			else
				--if  false then
					--结果正常 kv一般带有expire
					--local ttl_ = instance:ttl( key );
					--if  ttl_ and ttl_[1] then
						--ttl_ = ttl_[1]
					--else
						--ttl_ = nil
					--end
					--self:set_shared_cache_ssdb( ssdb_name, key .. "_kv", ret, ttl_ );
				--end

				self:set_get_set_tj( "kv", "get_ssdb" )
			end
		else
			--SSDB错误 一般是超时
			log.day( "ssdb", "get_fail|" .. key .. "|ret_1_not_string|" .. table.tostring(ret_table_) );
			self:error_exit( "get err:" .. (err or "nil") .. table.tostring(ret_table_) );

			--严重出错一定退出
			log.error( "error_exit3 error" );
			ngx.exit(ngx.ERROR);
			return;
		end

	else
		-- 返回的一定是一个table 否则是出错
		log.day( "ssdb", "get_fail|" .. key .. "|ret_not_table|" );
		self:error_exit( "get err:" .. (err or "nil") );

		--严重出错一定退出
		log.error( "error_exit4 error" );
		ngx.exit(ngx.ERROR);
		return;
	end

	return ret;
end



-------------------------------------------------------
_M.save = function( self, ssdb_name, key, hash, value, err_call_back, ssdb_delay_ )
	--if  _SSDB_CALL_LOG_ then
		--log.day_list_short( "ssdb_call",  ngx.ctx.m_params.act, ngx.ctx.m_params.cmd, "hset", ssdb_name, key, hash  );
	--end
	log.debug( "call ssdbMgr.save, ssdb_name=" .. ssdb_name .. ", key=" .. key .. ", hash=" .. hash );
	ngx.ctx.err_call_back = err_call_back;

	local key_hash_ = key .. "_h" .. hash

	--乐观锁w
	local instance

	if  ngx.ctx.__READ_ONLY__ == 1 then
		--只读接口有写操作
		log.day_list_simple( "lock_rw", "error_write", ngx.worker.id(),  uu.get_act_cmd(), "_v_" .. key_hash_, debug.traceback() );
	end

	--优先写入share cache
	local ret_ = self:set_shared_cache_ssdb( ssdb_name, key_hash_, value )
	if  ret_ and ssdb_delay_ and (tonumber(ssdb_delay_) == 1) then
		--log.day_list_short( "ssdb_delay", ssdb_name, key, hash )
		self:unlock_resty_lock( key_hash_ )
		self:set_get_set_tj( hash, "delay_ssdb" )
		return   --缓写
	end

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:read_unlock2(ssdb_name)
		self:error_exit( "write_lock_err" );
		return
	end

	if  not instance then
		instance = self:try_connect(ssdb_name);
	end
	local ok, err  = instance:hset(key, hash, value);
	self:read_unlock2(ssdb_name)

	if  not ok then
		log.day( "ssdb",  "hset_fail|" .. ssdb_name .. "|" .. key .. "|" .. hash .. "|" .. (err or "nil") );
		self:error_exit( "hset err:" .. (err or "nil") );
	else
		self:close()
		self:unlock_resty_lock( key_hash_ )
		--log.day( "ssdb",  "hset_succ|" .. ssdb_name .. "|" .. key .. "|" .. hash .. "|" .. #value );
	end

	self:set_get_set_tj( hash, "set_ssdb" )
end



_M.unlock_resty_lock = function ( self, pkey_hash_ )

	--解锁写
	if  _SSDB_RESTY_LOCK then
		if  ngx.ctx.resty_locks and ngx.ctx.resty_locks[ pkey_hash_ ] then

			if  ngx.ctx.resty_locks[ pkey_hash_ ].cc > 0 then
				if  ngx.ctx.resty_locks[ pkey_hash_ ].dict then
					local ok, err_ = ngx.ctx.resty_locks[ pkey_hash_ ].dict:delete( pkey_hash_ .. "_lk" )
					if  ok then
						--log.day_list_simple( "lock_rw", "unlock_ok1",    ngx.worker.id(), uu.get_act_cmd(), ngx.ctx.resty_locks[ pkey_hash_ ].cc, pkey_hash_, debug.traceback() );
					else
						log.day_list_simple( "lock_rw", "error_unlock1", ngx.worker.id(), uu.get_act_cmd(), ngx.ctx.resty_locks[ pkey_hash_ ].cc, pkey_hash_, err_, debug.traceback() );
					end
					ngx.ctx.resty_locks[ pkey_hash_ ].cc = 0  --解锁成功
				else
					--不加锁直接写,不用解开
				end
			else
				--重复解锁 使用缓存连写两次
				log.day_list_simple( "lock_rw", "warn_re_unlock1", ngx.worker.id(), uu.get_act_cmd(), 0, pkey_hash_, debug.traceback() );
			end

		else
			log.day_list_simple( "lock_rw", "error_unlock_no1", ngx.worker.id(), uu.get_act_cmd(), pkey_hash_, debug.traceback() );
		end
	end
end




--semaphore lock ( 相同worker下的不同协程同步 避免使用同一个socket句柄 )
_M.read_lock2 = function( self, pdb_name_ )

	do return 1 end  --haima todo

	local db_name_ = self.pre .. pdb_name_

	if  not self.read_semas_tj[db_name_] then
		self.read_semas_tj[db_name_] = {
			sema    = ns_semaphore.new(0),
			lock    = 0,
			lock_cc = 0,
			err_cc  = 0,
		}
	end

	local db_sema_ = self.read_semas_tj[db_name_];
	local start_   = ngx_now();
	local cc_      = 0;


	while true do
		cc_ = cc_ + 1;  --when wake up by post, other lthread getset lock=1, so this lthread continue wait

		if  db_sema_.lock == 0 then
			db_sema_.lock = 1   	-- got lock

			if  cc_ >= 5 then
				--get lock after >5 while-loops, try-lock is very busy and make lag
				log.day_list_short( "sema", "w",      db_name_, cc_,  math.floor((ngx_now()-start_)*1000) )
			end

			return 1

		else
			db_sema_.lock_cc = db_sema_.lock_cc + 1
			local ok, err = db_sema_.sema:wait(10);
			if  err then
				-- may timeout
				db_sema_.err_cc = db_sema_.err_cc + 1
				log.day_list( "sema", "error_" .. err, db_name_, cc_, db_sema_.lock_cc, db_sema_.err_cc, math.floor((ngx_now()-start_)*1000) )
				log.error("ERROR: read_semaphore " .. err)
				return 0
			end

		end
	end

	log.day_list( "sema", "error_while_break", db_name_, cc_,  math.floor((ngx_now()-start_)*1000) )
	return 0;
end


--semaphore unlock
_M.read_unlock2 = function( self, pdb_name_ )

	do return end  --haima todo

	local db_name_ = self.pre .. pdb_name_

	local db_sema_ = self.read_semas_tj[db_name_];
	if  db_sema_ then
		--test wait
		--ngx.sleep(1)
		db_sema_.lock = 0;          -- release lock

		if  db_sema_.sema:count() < 0 then
			db_sema_.sema:post(1)   --try awake a new wait thread
			db_sema_.lock_cc = db_sema_.lock_cc - 1

			local sema_count_ = db_sema_.sema:count()
			if  db_sema_.lock_cc % 16 == 15  or db_sema_.err_cc ~= 0 or
					( sema_count_ + db_sema_.lock_cc > 100 )   then
				--ssdb busy, too many try-lock; error; lock_cc too bigger than count
				log.day_list_short( "sema", "c", db_name_, sema_count_, db_sema_.lock_cc, db_sema_.err_cc )
				if  db_sema_.err_cc > 0 then
					db_sema_.err_cc = 0;
				end
			end

			if  sema_count_ == 0 then
				---对lock_cc进行逐步修正
				if  db_sema_.lock_cc > 0 then
					db_sema_.lock_cc = db_sema_.lock_cc - 1
				elseif  db_sema_.lock_cc < 0 then
					db_sema_.lock_cc = db_sema_.lock_cc + 1
				end
			end

		end
	end
end



_M.read = function( self, ssdb_name, key, hash, err_call_back )
	--if  _SSDB_CALL_LOG_ then
		--log.day_list_short( "ssdb_call",  ngx.ctx.m_params.act, ngx.ctx.m_params.cmd, "hget", ssdb_name, key, hash );
	--end
	log.debug( "call ssdbMgr.read, ssdb_name=" .. ssdb_name .. ", key=" .. key .. ", hash=" .. hash );
	ngx.ctx.err_call_back = err_call_back;

	self:set_get_set_tj( hash, "get_call" )

	local key_hash_ = key .. "_h" .. hash

	--乐观锁r
	local instance
	--ngx.ctx.__SSDB_READ__ = 1

	if  ngx.ctx.__READ_ONLY__ == 1 then
		--只读接口
	else
		if  _SSDB_RESTY_LOCK then
			local dict_ = ngx_shared[ "ssdb" .. ssdb_name ]
			if  dict_ then
				if  not ngx.ctx.resty_locks then
					ngx.ctx.resty_locks = {}
				end

				if  ngx.ctx.resty_locks[ key_hash_ ] then
					--已经锁定同样的key
					ngx.ctx.resty_locks[ key_hash_ ].cc = ngx.ctx.resty_locks[ key_hash_ ].cc + 1
					--log.day_list_simple( "lock_rw", "relock_add_0", ngx.worker.id(), uu.get_act_cmd(), key_hash_,
										--ngx.ctx.resty_locks[ key_hash_ ].cc, debug.traceback() );
				else

					local timeout_ = 2
					local step_    = 0.001
					while timeout_ > 0 do
						local ok, err = dict_:add( key_hash_ .. "_lk" , 1, 3)
						if  ok then
							--成功
							if  ngx.ctx.resty_locks[ key_hash_ ] then
								ngx.ctx.resty_locks[ key_hash_ ].cc = ngx.ctx.resty_locks[ key_hash_ ].cc + 1
								log.day_list_simple( "lock_rw", "error_relock0", ngx.worker.id(), uu.get_act_cmd(), key_hash_, 
								         ngx.ctx.resty_locks[ key_hash_ ].cc, debug.traceback() );
							else
								ngx.ctx.resty_locks[ key_hash_ ] = { cc=1, dict=dict_ }
							end

							if  step_ > 0.015 then   --1+2+4+8
								--循环等待其他锁的时间太长
								log.day_list_simple( "lock_rw", "lock_ok0", ngx.worker.id(), uu.get_act_cmd(), key_hash_, 
								                     ngx.ctx.resty_locks[ key_hash_ ].cc, timeout_, step_, debug.traceback() );
							end
							break

						else
							if  err == 'exists' then
								timeout_ = timeout_ - step_								
								if  timeout_ <= 0 then
									log.day_list_simple( "lock_rw", "error_time_out0", ngx.worker.id(), uu.get_act_cmd(), key_hash_, 
									                    timeout_, step_, debug.traceback() )
									break
								end

								if  step_ < 0.1 then
									step_ = step_ * 2
								end
								--log.day_list_simple( "lock_rw", "step0", ngx.worker.id(), uu.get_act_cmd(), key_hash_, timeout_, step_, debug.traceback() );
								ngx_sleep(step_)

							else
								log.day_list_simple( "lock_rw", "error_step0", ngx.worker.id(), uu.get_act_cmd(), key_hash_, 
								                    timeout_, step_, err, debug.traceback() );
								break

							end
						end
					end

				end

			else
				log.day_list_simple( "lock_rw", "error_no_ssdb_name", ngx.worker.id(), uu.get_act_cmd(), key_hash_, "ssdb" .. ssdb_name, debug.traceback() );
			end
		end

	end


	local ret2, stale_ = self:get_shared_cache_ssdb( ssdb_name, key_hash_ )
	if  (not stale_) and ret2 then
		log.debug( "ret2 cache hit." )
		self:set_get_set_tj( hash, "get_cache" )
		return ret2   --cache命中
	end

	if  stale_ then
		self:set_get_set_tj( hash, "get_cache_stale" )
	end

	local ret = ngx.null;

	--lock 模式
	local lock_id_ = self:read_lock2(ssdb_name)
	if  lock_id_  == 0 then
		self:read_unlock2(ssdb_name)
		self:error_exit( "read_lock_err" );
		return
	end

	if  not instance then
		instance = self:try_connect(ssdb_name);
	end
	local ret_table_, err = instance:hget(key, hash);
	self:read_unlock2(ssdb_name)

	if  err then
		log.day( "ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	else
		self:close()
	end

	uu.var_dump( ret_table_ );

	-- 处理返回值 返回是字符串或者ngx.null
	if  type(ret_table_) == 'table' then
		ret = ret_table_[1];
		if  ret and type(ret) == 'string' then
			if  'not_found' == ret then   --ssdb not_found
				log.debug( "ssdb ret not_found" );
				--log.day( "ssdb", "hget_succ|" .. key .. "|" .. hash  .. "|empty" );
				--结果为空
				ret = ngx.null;
			else
				--结果正常
				--log.day( "ssdb", "hget_succ|" .. key .. "|" .. hash  .. "|" .. #ret );

				--select_, key_, str_
				self:set_get_set_tj( hash, "get_ssdb" )
				self:set_shared_cache_ssdb( ssdb_name, key_hash_, ret )
			end
		else
			--SSDB错误 一般是超时
			log.day( "ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|ret_1_not_string|" .. table.tostring(ret_table_) );
			self:error_exit( "hget err:" .. (err or "nil") .. table.tostring(ret_table_) );

			--严重出错一定退出
			log.error( "error_exit1 error" );
			ngx.exit(ngx.ERROR);
			return;
		end

	else
		-- 返回的一定是一个table 否则是出错
		ret = ngx.null;
		log.day( "ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|ret_not_table|" );
		self:error_exit( "hget err:" .. (err or "nil") );

		--严重出错一定退出
		log.error( "error_exit2 error" );
		ngx.exit(ngx.ERROR);
		return;
	end

	return ret;
end



-- multi hget
_M.mread = function( self, ssdb_name, all_kv_list, err_call_back )
	log.debug( "call ssdbMgr.mread, ssdb_name=" .. ssdb_name );

	ngx.ctx.err_call_back = err_call_back;

	local fret = {};

	--先查cache
	local kv_list = {}           --需要ssdb的list
	for i=1, #all_kv_list do

		self.mread_tj.call = self.mread_tj.call + 1
		if  self.mread_tj.call >= 10000 then
			log.day_list( "cache_hit_ssdb", "mread", ssdb_name, self.mread_tj.call, self.mread_tj.get, self.mread_tj.set )
			self.mread_tj.call = 0
			self.mread_tj.get  = 0
			self.mread_tj.set  = 0
		end

		local key_hash_ = all_kv_list[i].key .. "_h" .. all_kv_list[i].hash
		local ret2, stale_ = self:get_shared_cache_ssdb( ssdb_name, key_hash_ )
		if (not stale_) and ret2 then
			--log.debug( "mread cache hit " .. key_hash_ )
			--log.day_list( "cache_hit_ssdb", "mm|cache_hit|" .. key_hash_ )
			fret[i] = ret2;
			self.mread_tj.get = self.mread_tj.get + 1
		else
			--log.debug( "mread need ssdb " .. key_hash_ )
			--log.day_list( "cache_hit_ssdb", "need_ssdb|" .. key_hash_ )
			--fret[i] = {}
			all_kv_list[i].__mi = i;     --mread index
			table_insert( kv_list, all_kv_list[i] )
		end
	end


	local ret_list = {};

	--lock 模式
	--local lock_id_ = self:read_lock2( ssdb_name )
	--if  lock_id_  == 0 then
		--self:read_unlock2(ssdb_name)
		--self:error_exit( "write_lock_err" );
		--return
	--end

	local instance = self:try_connect(ssdb_name);
	instance:init_pipeline();
	for i=1, #kv_list do
		instance:hget( kv_list[i].key, kv_list[i].hash );
	end
	local ret_table_list_, err = instance:commit_pipeline();

	--self:read_unlock2(ssdb_name)

	if  err then
		log.day( "ssdb", "mread|" .. (err or "nil") );
		self:error_exit( "mread err:" .. (err or "nil") );
		return;
	else
		self:close()
	end

	--uu.var_dump( ret_table_list_ );

	if type(ret_table_list_) == 'table' then

		for i=1, #ret_table_list_ do
			local ret_table_ = ret_table_list_[i];
			local ret = ngx.null;

			local key_ = "nil";
			if  kv_list[i] then
				key_ = kv_list[i].key or "nil";
			end

			-- 处理返回值 返回是字符串或者ngx.null
			if  type(ret_table_) == 'table' then
				ret = ret_table_[1];
				if  ret and type(ret) == 'string' then
					if  'not_found' == ret then   --ssdb not_found
						log.debug( "ssdb ret not_found|" .. key_ );
						--log.day( "ssdb", "hget_succ|empty" );
						ret = ngx.null;
					else
						--log.day( "ssdb", "hget_succ|" .. #ret );
						local key_hash_ = key_ .. "_h" .. kv_list[i].hash
						self:set_shared_cache_ssdb( ssdb_name, key_hash_, ret, 10*60 )
						self.mread_tj.set = self.mread_tj.set + 1
					end
				else
					ret = ngx.null;
					log.day( "ssdb", "mread|ret_1_not_string|" .. key_ );
				end

			else
				-- 返回的一定是一个table 否则是出错
				ret = ngx.null;
				log.day( "ssdb", "mread|ret_not_table|" .. key_ );
			end

			table_insert(ret_list, ret);
		end
	end


	--合并结果
	for i=1, #ret_list do
		if  kv_list[i] and kv_list[i].__mi then
			fret[ kv_list[i].__mi ] = ret_list[i]
		end
	end

	if  #fret ~= #all_kv_list then
		log.error("mread ret number not match")
		log.day_list("cache_hit_ssdb", "error_count")
	end

	return fret;

end



-- 删除一个key下的一个hash
_M.del = function( self, ssdb_name, key, hash, err_call_back )
	--if  _SSDB_CALL_LOG_ then
		--log.day_list_short( "ssdb_call",  ngx.ctx.m_params.act, ngx.ctx.m_params.cmd, "hdel", ssdb_name, key, hash );
	--end

	log.debug( "call ssdbMgr.del, ssdb_name=" .. ssdb_name .. ", key=" .. key .. ", hash=" .. hash );
	ngx.ctx.err_call_back = err_call_back;

	--self, select_, key_, hash_ )
	self:clean_shared_cache_ssdb( ssdb_name, key, hash )


	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:read_unlock2(ssdb_name)
		self:error_exit( "write_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hdel(key, hash);
	self:read_unlock2(ssdb_name)

	if  err then
		log.day( "ssdb", "hdel_fail|" .. ssdb_name .. "|" .. key .. "|" .. hash  .. "|" .. (err or "nil") );
		self:error_exit( "hdel err:" .. (err or "nil") );
		return
	else
		self:close()
	end

	self:unlock_resty_lock( ssdb_name, key .. "_h" .. hash )   --key_hash_

	uu.var_dump( ret_table_ );

	if  ret_table_ and ret_table_[1] then
		return ret_table_[1];
	else
		return 0;
	end
end


-- 一个key下有多少个hash值
_M.hlen = function( self, ssdb_name, key, err_call_back)
	--if  _SSDB_CALL_LOG_ then
		--log.day_list_short( "ssdb_call",  ngx.ctx.m_params.act, ngx.ctx.m_params.cmd, "hsize", ssdb_name, key );
	--end

	log.debug( "call ssdbMgr.hlen, ssdb_name=" .. ssdb_name .. ", key=" .. key );
	ngx.ctx.err_call_back = err_call_back;

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:read_unlock2(ssdb_name)
		self:error_exit( "write_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hsize(key);
	self:read_unlock2(ssdb_name)

	if  err then
		log.day( "ssdb", "hlen_fail|" .. ssdb_name .. "|" .. key  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	else
		self:close()
	end

	uu.var_dump( ret_table_ );

	if  ret_table_ and ret_table_[1] then
		return ret_table_[1];
	else
		return 0;
	end
end


--获得一个key下的所有值
_M.hget_all = function( self, ssdb_name, key, err_call_back )
	--if  _SSDB_CALL_LOG_ then
		--log.day_list_short( "ssdb_call",  ngx.ctx.m_params.act, ngx.ctx.m_params.cmd, "hgetall", ssdb_name, key );
	--end

	log.debug( "call ns_ssdbMgr.hget_all, key=" .. key );
	ngx.ctx.err_call_back = err_call_back;

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:read_unlock2(ssdb_name)
		self:error_exit( "write_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hgetall(key);
	self:read_unlock2(ssdb_name)

	if  err then
		log.day( "ssdb", "hlen_fail|" .. ssdb_name .. "|" .. key .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	else
		self:close()
	end

	uu.var_dump( ret_table_ );
	return ret_table_ or {};
end


-- 出错处理
_M.error_exit = function(self, txt)
	self:close(true)
	if  ngx.ctx.err_call_back then
		log.error( "has set error_exit." );
		ngx.ctx.err_call_back(txt);
	else
		log.error( "not set error_exit." );
	end
end


-- 关闭SSDB连接
_M.close = function( self, shutdown )
	if  ngx.ctx.last_ssdb_instance then
		self:private_close( ngx.ctx.last_ssdb_instance, shutdown );
		ngx.ctx.last_ssdb_instance = nil
	end
	--log.debug( "ssdb keep or close all ok." );
end


-- 关闭单个SSDB连接
_M.private_close = function( self, instance, shutdown )
	local ok, err;
	if  shutdown == true then
		ok, err = instance:close();
		log.day_list( "ssdb",  "close", self.pre );
	else
		ok, err = instance:set_keepalive( 120000, 256 );
	end

	if  not ok then
    	log.error("failed to set connection keepalive: " .. err .. ", pre=" .. self.pre );
    else
    	log.debug("connection keep alive ok. pre=" .. self.pre );
    end
end



-- 按连接名字获得ssdb配置
_M.get_ssdb_config = function( self, ssdb_name_ )
	if  self.config_cache[ ssdb_name_ ]  then
		--先查config cache
		log.debug( "find config_cache for " .. ssdb_name_ );
		return  self.config_cache[ ssdb_name_ ];
	else
		--if  ns_config and ns_config.ssdb_db_select_s2 and ns_config.ssdb_server_s2 then
		--config_server = config_server_;
		--config_select = config_select_;
		if  self.config_server and self.config_select then
			if  self.config_select[ssdb_name_] then
				local ret_ = self.config_server[ self.config_select[ssdb_name_] ];
				if  ret_ then
					log.debug( "set config_cache for " .. ssdb_name_ );
					self.config_cache[ ssdb_name_ ] = ret_;
					return ret_;
				end
			end
		end
	end

	self:error_exit( 'error ssdb_name = ' .. (ssdb_name_ or 'nil') );
end


-- 按照select_name_, 获得SSDB连接
-- ssdb_name_ = comm, uin,
_M.try_connect = function( self, ssdb_name_ )
	--self:init_pool();

	local config_ = self:get_ssdb_config( ssdb_name_ );

	--新建连接
	log.debug( "create new ssdb connection. " );
	local instance = self:private_try_connect( config_.ip, config_.port, config_.auth, ssdb_name_ );
	ngx.ctx.last_ssdb_instance = instance
	return instance

end


-----------------------------------------------------------------------------utils---------------------
-- config = { ip, port, auth, select }
-- 建立ssdb连接
_M.private_try_connect = function( self, ip, port, auth, ssdb_name_ )

	local instance = ns_ssdb:new();
	instance:set_timeout(3000);      --3s

	local ok, err = instance:connect( ip, port );
	if  not ok then
		self:read_unlock2(ssdb_name_)   		--- 唤醒下一个sema
		self:error_exit( "fail connect. msg=" .. err );
		return;
	end

	if  auth and #auth>0 then
		local count, err = instance:get_reused_times();
		if  0 == count then
			--set auth
			ok, err = instance:auth( auth );
			if  not ok then
				self:read_unlock2(ssdb_name_)   		--- 唤醒下一个sema
				self:error_exit( "fail auth. msg=" .. err );
				return;
			end

			ngx.ctx.ssdb_reused_count = 0;
		elseif err then
			self:read_unlock2(ssdb_name_)   		--- 唤醒下一个sema
			self:error_exit( "fail get_reused_times. msg=" .. err );
			return;
		else
			log.debug( "get_reused_times ssdb=" .. count );
			ngx.ctx.ssdb_reused_count = count;
		end
	end

	return instance;
end



--写校验[_k_] [_h_]
_M.private_check_set_key = function(self, db_name_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			log.day_list_short( "ssdb", "error_key_check", "set", db_name_, key, value._k_, hash, debug.traceback() );
			log.error( "error_set_key_check" )
			return false;
		end
	else
		value._k_ = tonumber(key) or key;  --首次设置
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			log.day_list_short( "ssdb", "error_hash_check", "set", db_name_, key, value._h_, hash );
			log.error( "error_set_hash_check" )
			return false;
		end
	else
		if  tonumber(hash) then
			value._h_ = tonumber(hash) --首次设置
		end
	end

	value._t_ = uu.now()    --最后一次写入时间
	return true;
end



--读校验[_k_] [_h_]
_M.private_check_get_key = function(self, db_name_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			--读取出错
			log.day_list_short( "ssdb", "error_key_check", "get", db_name_, key, value._k_, hash, debug.traceback() );
			log.error( "error_get_key_check" )
			return false;
		end
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			--读取出错
			log.day_list_short( "ssdb", "error_hash_check", "get", db_name_, key, value._h_, hash );
			log.error( "error_get_hash_check" )
			return false;
		end
	end

	return true;
end





-----------------------------ngx_shared--------------------------------




--设置全局数据缓存kv 十分表
_M.set_shared_cache_ssdb = function( self, select_, key_, str_, expire_ )
	if  select_ and key_ and str_  then

		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = "ssdb" .. (select_%10)
		else
			dict_name_ = "ssdb" ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			if  expire_ and type(expire_) == 'string' then
				--log.debug( "expire_=" .. expire_ .. " k=" .. key_ .. " v=" .. str_ )
				expire_ = tonumber(expire_) or 0
			end

			if  expire_ and expire_ >=0 and expire_ < 1800 then
				--normal
			else
				expire_ = 1800     --30*60
			end

			local ok, err, forcible = share_:set( key_, str_, expire_ );
			if  ok then
				if  forcible then
				    log.day_list_short( "cache_hit_ssdb", "set_forcible", dict_name_, key_, err )
				end
				return true
			else
				if  err then
					--share_:delete( key_ )
					--set出错，立即删除缓存
					local ok, err, forcible = share_:set( key_, nil );
					if  err or forcible then
						log.day_list_short( "cache_hit_ssdb", "error_delete", dict_name_, key_, err, forcible )
					end
				end
				log.day_list_short( "cache_hit_ssdb", "error_set", dict_name_, key_, err )
			end
		else
			log.day_list_short( "cache_hit_ssdb", "error_set_no_dict", dict_name_, key_)
		end
	end

	return false   --写入cache失败
end



--获得全局数据缓存kv 十分表
_M.get_shared_cache_ssdb = function( self, select_, key_ )
	if  select_ and key_ then

		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = "ssdb" .. (select_%10)
		else
			dict_name_ = "ssdb" ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			local str_, f_, stale_ = share_:get_stale( key_);
			--uu.report_share_free_space( share_, dict_name_, key_, "cache_hit_ssdb" )

			if  str_ then
				return str_, stale_  	--正常返回
			else
				--有报错的情况 ret = nil, error
				if  f_ then
					log.day_list_short( "cache_hit_ssdb", "error_get_nil", dict_name_, key_, f_ )
				end
			end
		else
			log.day_list_short( "cache_hit_ssdb", "error_get_no_dict", dict_name_, key_)
		end
	end

	return nil, nil
end



--清理掉一个cache数据 十分表
_M.clean_shared_cache_ssdb = function(self, select_, key_, hash_ )
	if  select_ and key_ and hash_ then
		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = "ssdb" .. (select_%10)
		else
			dict_name_ = "ssdb" ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then
			local ok, err, forcible = share_:set( key_ .. "_h" .. hash_ , nil );
			if  err or forcible then
				log.day_list_short( "cache_hit_ssdb", "error_delete", dict_name_, key_, err, forcible )
			end
		end
	end
end



return _M;
